package com.sdhs.paas.flux.app;

import java.util.Objects;

import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.function.BodyExtractors;
import org.springframework.web.reactive.function.client.WebClient;

@RestController
public class SSEClientController {
	
	@GetMapping("/sseclient")
	public String sseClient() {
		final WebClient client = WebClient.create();
		client.get().uri("http://localhost:8087/sse/randomNumbers").accept(MediaType.TEXT_EVENT_STREAM).exchange()
				.flatMapMany(response -> response
						.body(BodyExtractors.toFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {
						})))
				.filter(sse -> Objects.nonNull(sse.data())).map(ServerSentEvent::data).buffer(10)
				.doOnNext(System.out::println).blockFirst();
		return "sseclient";
	}

	public static void main(final String[] args) {
		final WebClient client = WebClient.create();
		client.get().uri("http://localhost:8087/sse/randomNumbers").accept(MediaType.TEXT_EVENT_STREAM).exchange()
				.flatMapMany(response -> response
						.body(BodyExtractors.toFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {
						})))
				.filter(sse -> Objects.nonNull(sse.data())).map(ServerSentEvent::data).buffer(10)
				.doOnNext(System.out::println).blockFirst();
	}
}
